package net.xdclass.dwm;

import lombok.extern.slf4j.Slf4j;
import net.xdclass.func.AsyncLocaltionRequestFunction;
import net.xdclass.func.DeviceMapFunction;
import net.xdclass.model.ShortLinkWideDO;
import net.xdclass.util.KafkaUtil;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.concurrent.TimeUnit;

@Slf4j
public class DwmShortLinkWideApp {

    private static final String SOURCE_TOPIC = "dwd_link_visit_topic";

    private static final String SINK_TOPIC = "dwm_short_link_group";

    private static final String GROUP_ID = "dwm_short_link_group";

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        // 1 获取流
        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(SOURCE_TOPIC, GROUP_ID);

        DataStreamSource<String> ds = env.addSource(kafkaConsumer);

        ds.print();
        //2 补全设备信息
        SingleOutputStreamOperator<ShortLinkWideDO> deviceWideDS = ds.map(new DeviceMapFunction());

        deviceWideDS.print("设备信息宽表补齐");

        //3 补全地理位置信息
//        SingleOutputStreamOperator<String> shortLinkWideDS = deviceWideDS.map(new LocationMapFunction());
        SingleOutputStreamOperator<String> shortLinkWideDS =
                AsyncDataStream.unorderedWait(deviceWideDS, new AsyncLocaltionRequestFunction(), 100, TimeUnit.MILLISECONDS, 200);



        FlinkKafkaProducer<String> kafkaProducer = KafkaUtil.getKafkaProducer(SINK_TOPIC);

        //4 将sink写到dwm层 kafka存储
        shortLinkWideDS.addSink(kafkaProducer);

        env.execute();

    }
}
